package demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;

@Slf4j
public class RatingsUsersMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    private HashMap<String, String> user_info = new HashMap<String, String>();
    private String splitter = "";
    private String rating_secondPart = "";

    public static String userFile = "";

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        splitter = context.getConfiguration().get("SPLITTER");
		/*Path[] distributePaths=DistributedCache.getLocalCacheFiles(context.getConfiguration());
		String line="";
		BufferedReader br=null;
		for (Path path : distributePaths) {
			if(path.toString().endsWith("users.dat")){
				//TODO:由于无法读取到DistributedCache方法配置的文件，所以采用下面的方式进行
				//上传文件：hdfs dfs -put /root/myhadoop/data/movie/* /movie
				//查看文件：hadoop fs -ls /movie
				//授权文件：hdfs dfs -chmod 744 /movie/users.dat
				br=new BufferedReader(new FileReader(path.toString()));
				while((line=br.readLine())!=null){
					String userID=line.substring(0, line.indexOf("::"));
					String secondPart=line.substring(line.indexOf("::")+2,line.length());
					user_info.put(userID, secondPart);
				}
			}
		}*/

        Configuration conf = context.getConfiguration();
        log.info("配置路径：{}", conf.get("fs.defaultFS"));
        FileSystem fs = FileSystem.get(conf);
        //TODO:在实际的生产环境中，此userFile变量无法动态获取到值
        Path filePath = new Path(StringUtils.isEmpty(userFile) ? "/movie/users.dat" : userFile); // HDFS文件路径
        try (BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(filePath)))) {
            String line;
            while ((line = br.readLine()) != null) {
                //users  1::F::1::10::48067
                String userID = line.substring(0, line.indexOf("::"));
                String secondPart = line.substring(line.indexOf("::") + 2, line.length());
                user_info.put(userID, secondPart);
            }
        }
//        fs.close();//注意：此处不能关闭
    }


    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context)
            throws IOException, InterruptedException {
        //ratings 1::1193::5::978300760
        String[] val = value.toString().split(splitter);//解析的是 用户对电影的评分数据ratings.dat
        rating_secondPart = user_info.get(val[0]);
        if (rating_secondPart != null) {
            //一份包含UserID（用户ID），Gender（性别），Age（年龄），Occupation（职业），Zip-code（编码），MovieID（电影ID）的数据
            String result = val[0] + splitter + rating_secondPart + splitter + val[1];
            context.write(new Text(result), NullWritable.get());
        }
    }

}
